﻿using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Threading.Tasks;
using Samurai.Wakizashi;

namespace Reactor.Messaging.ServiceBus.Internals
{
    public partial class TaskDistributor : DisposableBase
    {
        #region Fields

        private readonly int _maxNumberOfConsumingSessions;
        private readonly BlockingCollection<TaskBase> _taskCollection;
        private readonly List<TaskProcessor> _consumptionProcessors;
        private readonly List<TaskProcessor> _transactionalProcessors;
        private int _nextConsumptionProcessorIndex;
        private readonly object _processorSync = new object();

        #endregion

        #region Constructors

        /// <summary>
        /// Initializes a new instance of the <see cref="TaskDistributor"/> class.
        /// </summary>
        /// <param name="nmsGateway">The JMS manager.</param>
        /// <param name="maxNumberOfConsumingSessions">The max number of consuming sessions.</param>
        public TaskDistributor(INmsGateway nmsGateway, int maxNumberOfConsumingSessions)
        {
            if (nmsGateway == null) throw new ArgumentNullException("nmsGateway");

            _maxNumberOfConsumingSessions = maxNumberOfConsumingSessions;
            _taskCollection = new BlockingCollection<TaskBase>();
            _consumptionProcessors = new List<TaskProcessor>(maxNumberOfConsumingSessions);
            _transactionalProcessors = new List<TaskProcessor>();
            TransmissionProcessor = new TaskProcessor(this, nmsGateway);
            TransmissionProcessor.MessageSent += TransmissionProcessor_MessageSent;
            TransmissionProcessor.Start();

            //Create consumption processors
            for (var i = 0; i < _maxNumberOfConsumingSessions; i++)
            {
                var processor = new TaskProcessor(this, nmsGateway);
                _consumptionProcessors.Add(processor);
                processor.MessageReceived += MessageReceiver_MessageReceived;
                processor.Start();
            }
        }

        protected void MessageReceiver_MessageReceived(object sender, MessageEventArgs e)
        {
            // Bubble MessageReceived event upward
            if (MessageReceived != null) MessageReceived(this, e);
        }

        #endregion

        /// <summary>
        /// Occurs when a message is sent by the TransmissionProcessor.
        /// </summary>
        public event EventHandler<MessageEventArgs> MessageSent;

        /// <summary>
        /// Occurs when a message is received.
        /// </summary>
        public event EventHandler<MessageEventArgs> MessageReceived;

        public void EnqueueTask(TaskBase task)
        {
            _taskCollection.Add(task);
        }

        public void Start()
        {
            // Begin processing tasks
            Task.Factory.StartNew(TakeTask);
        }

        public void Stop()
        {
            _taskCollection.CompleteAdding();

            DetachSentAndReceivedEvents();
        }

        public List<TaskProcessor> ConsumptionProcessors
        {
            get { return _consumptionProcessors; }
        }

        public TaskProcessor TransmissionProcessor { get; set; }

        /// <summary>
        /// Handles the MessageSent event of the TransmissionProcessor instance by in turn raising a MessageSent event.
        /// </summary>
        /// <param name="sender">The source of the event.</param>
        /// <param name="e">The <see cref="Reactor.Messaging.ServiceBus.MessageEventArgs"/> instance containing the event data.</param>
        protected virtual void TransmissionProcessor_MessageSent(object sender, MessageEventArgs e)
        {
            if (MessageSent != null) MessageSent(this, e);
        }

        private void TakeTask()
        {
            while (!_taskCollection.IsCompleted)
            {
                TaskBase task = null;
                try
                {
                    task = _taskCollection.Take();
                }
                catch (InvalidOperationException) { /* IOE means that Take() was called on a complete collection */ }

                if (task is SubscriptionTask)
                    Task.Factory.StartNew(() => HandOffToConsumptionProcessor((SubscriptionTask)task));
                else if (task is TransmissionTask)
                    Task.Factory.StartNew(() => TransmissionProcessor.ProcessTask(task));
            }

            KillProcessors();
        }

        private TaskProcessor GetProcessorServicingSpecificDestination(string typedDestinationName)
        {
            lock (_processorSync)
            {
                foreach (var consumptionProcessor in ConsumptionProcessors)
                {
                    if (consumptionProcessor.IsServicingDestination(typedDestinationName))
                        return consumptionProcessor;
                }
            }

            return null;
        }

        private void HandOffToConsumptionProcessor(SubscriptionTask task)
        {
            // See if we have a processor already servicing the intended MessagingChannel
            var processor = GetProcessorServicingSpecificDestination(task.MessagingChannel.GetTypedDestinationName());
            if(processor != null)
            {
                processor.ProcessTask(task);
                return;
            }

            // No processor servicing the intended MessagingChannel, so choose the next in line to do so
            if (_nextConsumptionProcessorIndex >= _maxNumberOfConsumingSessions)
                _nextConsumptionProcessorIndex = 0;

            processor = _consumptionProcessors[_nextConsumptionProcessorIndex];
            processor.ProcessTask(task);

            _nextConsumptionProcessorIndex++;
            if (_nextConsumptionProcessorIndex >= _maxNumberOfConsumingSessions)
                _nextConsumptionProcessorIndex = 0;
        }

        private void KillProcessors()
        {
            lock (_processorSync)
            {
                foreach (var consumptionProcessor in ConsumptionProcessors)
                {
                    consumptionProcessor.Stop();
                    consumptionProcessor.MessageReceived -= MessageReceiver_MessageReceived;
                }

                foreach(var transactionalProcessor in _transactionalProcessors)
                {
                    transactionalProcessor.Stop();
                    transactionalProcessor.MessageReceived -= MessageReceiver_MessageReceived;
                }

                if(TransmissionProcessor != null)
                {
                    TransmissionProcessor.Stop();
                    TransmissionProcessor.MessageSent -= TransmissionProcessor_MessageSent;
                }
            }
        }

        private void DetachSentAndReceivedEvents()
        {
            if (TransmissionProcessor != null)
                TransmissionProcessor.MessageSent -= TransmissionProcessor_MessageSent;

            foreach (var consumptionProcessor in ConsumptionProcessors)
                consumptionProcessor.MessageReceived -= MessageReceiver_MessageReceived;
        }

        protected override void Dispose(bool disposing)
        {
            if(_taskCollection != null) _taskCollection.Dispose();

            base.Dispose(disposing);
        }
    }
}
